package org.xwb;

import lombok.Data;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

@Data
class MyProgress extends SubmissionPublisher<String> implements Flow.Subscriber<String> {
    public MyProgress(String name) {
        super();
        this.name = name;
    }
    String name;
    Flow.Subscription subscription;
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("中间商"+name+"-开始订阅"+Thread.currentThread().getName()+"线程-"+subscription);
        this.subscription = subscription;
        subscription.request(1);
    }
    @Override
    public void onNext(String item) {
        System.out.println("中间商"+name+"-当前元素"+Thread.currentThread().getName()+"线程-"+item);
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        submit(name+"-"+item);
        subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("中间商"+name+"-订阅出错"+Thread.currentThread().getName()+"线程-"+throwable);
    }

    @Override
    public void onComplete() {
        System.out.println("中间商"+name+"-订阅完成"+Thread.currentThread().getName()+"线程-");
    }
}

public class FlowDemo {
    public static void main(String[] args) {
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        MyProgress a = new MyProgress("A号");
        MyProgress b = new MyProgress("B号");
        MyProgress c = new MyProgress("C号");
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {
            Flow.Subscription subscription;
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println("订阅者-开始订阅"+Thread.currentThread().getName()+"线程-"+subscription);
                this.subscription = subscription;
                subscription.request(1);
            }
            @Override
            public void onNext(String item) {
                System.out.println("订阅者-当前元素"+Thread.currentThread().getName()+"线程-"+item);
                subscription.request(1);
            }
            @Override
            public void onError(Throwable throwable) {
                System.out.println("订阅者-订阅出错"+Thread.currentThread().getName()+"线程-"+throwable);
            }
            @Override
            public void onComplete() {
                System.out.println("订阅者-订阅完成"+Thread.currentThread().getName()+"线程-");
            }
        };
        publisher.subscribe(a);
        a.subscribe(b);
        b.subscribe(c);
        c.subscribe(subscriber);




        for (int i = 0; i < 10; i++) {
            publisher.submit("第"+i+"个");
        }

        publisher.close();

        try {
            TimeUnit.SECONDS.sleep(20);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

}